package com.tqz.rc.server.consistency;

import com.tqz.rc.common.exception.RcException;
import com.tqz.rc.common.util.StringUtils;
import com.tqz.rc.server.core.Instances;
import javafx.util.Pair;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.concurrent.*;

/**
 * <p>一致性数据服务实现类，用来保证server端各个节点的数据同步、复制等操作。
 *
 * @author tianqingzhao
 * @since 2021/8/26 17:48
 */
@Component
public class DistroConsistencyServiceImpl implements EphemeralConsistencyService {

    private DataStore dataStore;

    private volatile Notifier notifier = new Notifier();

    private Map<String, ConcurrentLinkedQueue<RecordListener>> listeners = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        dataStore = new DataStore();
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        executorService.submit(notifier);
    }

    @Override
    public void put(String key, Record value) {
        onPut(key, value);

        // TODO 同步实例信息到集群其他节点
        // distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,
        //                DistroConfig.getInstance().getSyncDelayMillis());
    }

    @Override
    public Datum get(String key) {
        return dataStore.get(key);
    }

    @Override
    public void listen(String key, RecordListener recordListener) {
        if (!listeners.containsKey(key)) {
            listeners.put(key, new ConcurrentLinkedQueue<>());
        }

        if (listeners.get(key).contains(recordListener)) {
            return;
        }

        listeners.get(key).add(recordListener);
    }

    private void onPut(String key, Record value) {
        if (KeyBuilder.matchEphemeralInstanceListKey(key)) {
            Datum<Instances> datum = new Datum<>();
            datum.key = key;
            datum.value = (Instances) value;
            dataStore.put(key, datum);
        }

        if (!listeners.containsKey(key)) {
            return;
        }

        notifier.addTask(key, DataOperation.CHANGE);
    }

    public class Notifier implements Runnable {

        private ConcurrentHashMap<String, String> services = new ConcurrentHashMap<>(10 * 1024);

        private BlockingQueue<Pair<String, DataOperation>> tasks = new ArrayBlockingQueue<>(1024 * 1024);

        public void addTask(String key, DataOperation action) {
            if (services.containsKey(key) && action == DataOperation.CHANGE) {
                return;
            }

            if (action == DataOperation.CHANGE) {
                services.put(key, StringUtils.EMPTY);
            }

            tasks.offer(new Pair(key, action));
        }

        @Override
        public void run() {
            System.out.println("distro notifier start.");

            for (; ; ) {
                try {
                    Pair<String, DataOperation> pair = tasks.take();
                    handle(pair);
                } catch (InterruptedException | RcException e) {
                    e.printStackTrace();
                }
            }
        }

        /**
         * 执行服务的变更。
         *
         * @param pair
         */
        private void handle(Pair<String, DataOperation> pair) throws RcException {
            String key = pair.getKey();
            DataOperation action = pair.getValue();

            services.remove(key);

            if (!listeners.containsKey(key)) {
                return;
            }

            int count = 0;

            for (RecordListener listener : listeners.get(key)) {

                count++;

                if (action == DataOperation.CHANGE) {
                    listener.onChange(key, dataStore.get(key).value);
                    continue;
                }

            }

            System.out.println("[TqzRc-DISTRO] datum change notified, key: " + key +
                    ", listener count: " + count+ ", action: "+ action);
        }
    }
}
